package com.lvmama.rhino.common.utils.spark

import org.apache.spark.SparkContext
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
  * Created by yuanxiaofeng on 2016/6/23.
  */
trait SparkStreamingApplication extends SparkApplication{
  def streamingBatchDuration: Long

  def withSparkStreamingContext(f: (SparkContext, StreamingContext) => Unit): Unit = {
    withSparkContext { sc =>
      val ssc = new StreamingContext(sc, Seconds(streamingBatchDuration))
//      ssc.checkpoint(streamingCheckpointDir)
      f(sc, ssc)

      ssc.start()
      ssc.awaitTermination()
    }
  }

}
